package com.yq.canalkafka.service.impl;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.common.base.Strings;
import com.yq.canalkafka.config.StaticConfig;
import com.yq.canalkafka.service.CanalSqlService;
import com.yq.canalkafka.util.CanalKafkaUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @author by shizan
 * @Classname CanalSqlServiceImpl
 * @Description TODO
 * @Date 2020/11/23 4:36 下午
 */
@Service
@Slf4j
public class CanalSqlServiceImpl implements CanalSqlService {
    @Value("${canal.from-type}")
    private Integer fromType;

    @Override
    public String getInsertSql(List<CanalEntry.Column> columnsList, String tableName) {
//        StringBuilder sqlBu = new StringBuilder("insert into `" + tableName + "` (");
//        StringBuilder sqlKeysBu = new StringBuilder("");
//        StringBuilder sqlValuesBu = new StringBuilder(") VALUES(");
        String sql = "insert into `" + tableName + "` (";
        String sqlKeys = "";
        String sqlValues = ") VALUES(";
        List<CanalEntry.Column> columnsNew = null;
        if (!CollectionUtils.isEmpty(columnsList)) {
            columnsNew = columnsList.stream().filter(c -> c.getUpdated() == true && c.getIsNull() == false).collect(Collectors.toList());
        }

        if (CollectionUtils.isEmpty(columnsNew)) {
            return null;
        }

//        for (CanalEntry.Column column : columnsList) {
//            //from_type存在 且与 接收到的from_type与配置的相同时，sql才需要执行
//            if (column.getName().equals(StaticConfig.CHANGE_COLUMN)
//                    && !Strings.isNullOrEmpty(column.getName())) {
//                Integer val = Integer.valueOf(column.getValue());
//                if (val.equals(fromType)) {
//                    flag = true;
//
//                }
//
//
//            }
//        }
        boolean flag = false;
        int j=0;
        for (int i = 0; i < columnsList.size(); i++) {
            CanalEntry.Column column = columnsList.get(i);
            if (column.getName().equals(StaticConfig.CHANGE_COLUMN)
                    && !Strings.isNullOrEmpty(column.getName())) {
                Integer val = Integer.valueOf(column.getValue());
                if (val.equals(fromType)) {
                    //from_type存在 且与 接收到的from_type与配置的相同时，sql才需要执行
                    flag = true;
                }

            }
            if (j == 0) {
//                sqlKeysBu.append(sqlKeysBu).append("`" + column.getName() + "`");
//                sqlValuesBu.append(sqlValuesBu).append(getValue(column));
                sqlKeys = sqlKeys + "`" + column.getName() + "`";
                sqlValues = sqlValues + getValue(column);
                j++;
            } else {
//                sqlKeysBu.append(sqlKeysBu).append(" ,`" + column.getName() + "`");
//                sqlValuesBu.append(sqlValuesBu).append("," + getValue(column));
                sqlKeys = sqlKeys + " ,`" + column.getName() + "`";
                sqlValues = sqlValues + "," + getValue(column);
                j++;
            }
        }
        //没有from_type的话，直接不复制。
        if (!flag) {
            log.info("无需新增原始语句");
//            CanalKafkaUtils.printColumn(columnsList);
            return null;
        }
        return sql + sqlKeys + sqlValues + ")";
//        return sqlBu.append(sqlKeysBu).append(sqlValuesBu).append(")").toString();
    }


    @Override
    public String getUpdateSql(List<CanalEntry.Column> columnsList, String tableName) {


        String sql = "update `" + tableName + "` set ";
        String where = " where ";
        List<CanalEntry.Column> columnsNew = null;
        if (!CollectionUtils.isEmpty(columnsList)) {
            columnsNew = columnsList.stream().filter(c -> (c.getUpdated() == true || c.getName().equals(StaticConfig.CHANGE_COLUMN))).collect(Collectors.toList());
        }

        if (CollectionUtils.isEmpty(columnsNew)) {
            return null;
        }

        boolean flag = false;

        for (CanalEntry.Column column : columnsList) {
            if (column.getName().equals(StaticConfig.CHANGE_COLUMN)) {
                Integer val = Integer.valueOf(column.getValue());
                if (val.equals(fromType)) {
                    flag = true;
                }

            }
            if (column.getIsKey()) {
                where = where + column.getName() + " = '" + column.getValue()+"'";
                continue;
            }
        }

       int j =0;
        for (int i = 0; i < columnsNew.size(); i++) {
            CanalEntry.Column column = columnsNew.get(i);
            if (!column.getName().equals(StaticConfig.CHANGE_COLUMN)) {
                //不需要更新from_type字段
                if (j == 0) {
                    sql = sql + "`" + column.getName() + "` = " + getValue(column);
                    j++;
                } else {
                    sql = sql + ", `" + column.getName() + "` = " + getValue(column);
                    j++;
                }
            }

        }
        if (flag) {
            return sql + where;
        } else {
            log.info("无需更新原始语句");
//            CanalKafkaUtils.printColumn(columnsList);
            return null;
        }

    }

    private String getValue(CanalEntry.Column column) {
        if (column.getIsNull()) {
            return "null";
        }
        return String.format("'%s'", column.getValue());
    }
}
